package com.djf.flinkcdc.service;

import com.djf.flinkcdc.config.FlinkCdcProperties;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.PrintSinkFunction;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * CDC 数据同步服务
 * 提供数据同步、数据一致性和数据迁移功能
 * 
 * 🔧 核心功能模块说明：
 * 
 * 1. 【数据同步模块】
 *    - 基于 Flink CDC 实现 MySQL binlog 实时同步
 *    - 支持全量+增量、仅增量、从最早开始等多种同步模式
 *    - 通过 MySqlSource 构建数据源，实现低延迟数据捕获
 * 
 * 2. 【数据一致性模块】
 *    - 采用 Flink 的 Exactly-Once 语义保证数据一致性
 *    - 通过 Checkpoint 机制实现故障恢复和状态管理
 *    - 数据处理器中集成一致性检查逻辑
 * 
 * 3. 【数据迁移模块】
 *    - 支持历史数据的全量迁移（initial 模式）
 *    - 支持实时数据的增量迁移（latest-offset 模式）
 *    - 通过流式处理实现大规模数据迁移
 * 
 * 4. 【环境配置模块】
 *    - 自动配置 Flink 执行环境和优化参数
 *    - 集成重启策略和状态后端配置
 *    - 支持 Web UI 监控界面
 * 
 * 5. 【监控管理模块】
 *    - 提供同步任务的启动、停止、状态查询功能
 *    - 集成详细的日志记录和错误处理
 *    - 支持并发安全的任务管理
 */
@Slf4j
@Service
public class CdcSyncService {

    @Autowired
    private FlinkCdcProperties properties;

    private StreamExecutionEnvironment env;
    private MySqlSource<String> mySqlSource;
    private final AtomicBoolean isRunning = new AtomicBoolean(false);
    private CompletableFuture<Void> syncTask;

    @PostConstruct
    public void init() {
        log.info("🔧 初始化 Flink CDC 同步服务...");
        setupFlinkEnvironment();
        buildMySqlSource();
        log.info("✅ Flink CDC 同步服务初始化完成");
    }

    /**
     * 🔧 配置 Flink 执行环境
     * 
     * 【环境配置段落】- 核心执行环境初始化
     * 此段落负责配置 Flink 流处理环境的所有核心参数：
     * 
     * 1. 执行环境创建：
     *    - 创建本地执行环境并启用 Web UI 监控
     *    - 支持通过浏览器实时查看任务执行状态
     * 
     * 2. 并行度设置：
     *    - 设置为 1 保证数据处理的顺序性
     *    - 避免并发处理导致的数据乱序问题
     * 
     * 3. 重启策略配置：
     *    - 固定延迟重启策略：最多重试 3 次，每次间隔 10 秒
     *    - 提供任务失败时的自动恢复能力
     * 
     * 4. Checkpoint 检查点配置：
     *    - Exactly-Once 语义保证数据一致性
     *    - 配置检查点间隔、超时、并发数等参数
     *    - 启用外部化检查点支持故障恢复
     * 
     * 5. 状态后端设置：
     *    - 使用文件系统状态后端持久化状态数据
     *    - 支持大规模状态存储和快速恢复
     */
    private void setupFlinkEnvironment() {
        Configuration config = new Configuration();
        env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(config);
        
        // 设置并行度
        env.setParallelism(1);
        
        // 配置重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.seconds(10)));
        
        // 配置 Checkpoint
        if (properties.getCheckpoint().getEnabled()) {
            env.enableCheckpointing(properties.getCheckpoint().getInterval());
            env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
            env.getCheckpointConfig().setCheckpointTimeout(60000);
            env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
            env.getCheckpointConfig().enableExternalizedCheckpoints(
                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
            
            // 设置状态后端
            env.setStateBackend(new FsStateBackend(properties.getCheckpoint().getStorage()));
        }
        
        log.info("🔧 Flink 环境配置完成 - 并行度: 1, Checkpoint: {}ms", 
                properties.getCheckpoint().getInterval());
    }

    /**
     * 📊 构建 MySQL CDC 源
     * 
     * 【数据源构建段落】- MySQL CDC 连接器配置
     * 此段落负责构建和配置 MySQL CDC 数据源：
     * 
     * 1. 连接配置：
     *    - 配置 MySQL 服务器连接信息（主机、端口、用户名、密码）
     *    - 设置数据库和表的监听范围
     * 
     * 2. 启动模式选择：
     *    - initial: 全量+增量同步，先读取存量数据再监听增量
     *    - latest-offset: 仅增量同步，只监听新的数据变更
     *    - earliest-offset: 从最早的 binlog 开始读取
     * 
     * 3. 服务器配置：
     *    - server-id: MySQL 复制的唯一标识符
     *    - server-time-zone: 时区设置，确保时间数据的正确性
     * 
     * 4. 序列化配置：
     *    - 使用 JsonDebeziumDeserializationSchema 将 CDC 数据转换为 JSON
     *    - 包含 schema 变更信息，支持表结构变更监听
     * 
     * 5. 数据格式：
     *    - 输出标准的 Debezium JSON 格式
     *    - 包含操作类型、变更前后数据、时间戳等完整信息
     */
    private void buildMySqlSource() {
        StartupOptions startupOptions = getStartupOptions();
        
        mySqlSource = MySqlSource.<String>builder()
                .hostname(properties.getMysql().getHostname())
                .port(properties.getMysql().getPort())
                .username(properties.getMysql().getUsername())
                .password(properties.getMysql().getPassword())
                .databaseList(properties.getMysql().getDatabaseList())
                .tableList(properties.getMysql().getTableList())
                .startupOptions(startupOptions)
                .serverId(properties.getMysql().getServerId())
                .serverTimeZone(properties.getMysql().getServerTimeZone())
                .deserializer(new JsonDebeziumDeserializationSchema())
                .includeSchemaChanges(true)
                .build();
        
        log.info("📊 MySQL CDC 源构建完成 - 数据库: {}, 表: {}, 模式: {}", 
                properties.getMysql().getDatabaseList(),
                properties.getMysql().getTableList(),
                properties.getMysql().getStartupMode());
    }

    /**
     * 获取启动选项
     */
    private StartupOptions getStartupOptions() {
        String mode = properties.getMysql().getStartupMode().toLowerCase();
        switch (mode) {
            case "initial":
                log.info("🔄 启动模式: INITIAL - 全量+增量同步");
                return StartupOptions.initial();
            case "latest-offset":
                log.info("⚡ 启动模式: LATEST - 仅增量同步");
                return StartupOptions.latest();
            case "earliest-offset":
                log.info("📚 启动模式: EARLIEST - 从最早开始");
                return StartupOptions.earliest();
            default:
                log.warn("⚠️ 未知启动模式: {}, 使用默认 INITIAL", mode);
                return StartupOptions.initial();
        }
    }

    /**
     * 🚀 启动数据同步
     * 
     * 【同步任务启动段落】- 核心数据同步流程
     * 此段落实现完整的数据同步任务启动和执行：
     * 
     * 1. 状态检查：
     *    - 检查同步任务是否已在运行，避免重复启动
     *    - 使用 AtomicBoolean 保证线程安全的状态管理
     * 
     * 2. 异步任务执行：
     *    - 使用 CompletableFuture 异步执行同步任务
     *    - 避免阻塞主线程，支持并发操作
     * 
     * 3. 数据流构建：
     *    - 从 MySQL CDC 源创建数据流
     *    - 设置 WatermarkStrategy 处理事件时间
     *    - 配置并行度确保数据顺序性
     * 
     * 4. 数据处理链：
     *    - 通过 DataProcessor 处理 CDC 数据
     *    - 实现数据解析、一致性检查、格式转换
     *    - 输出到控制台进行实时监控
     * 
     * 5. 任务执行：
     *    - 调用 env.execute() 启动 Flink 作业
     *    - 集成异常处理和状态管理
     *    - 提供详细的日志记录
     */
    public synchronized void startSync() {
        if (isRunning.get()) {
            log.warn("⚠️ 同步任务已在运行中");
            return;
        }
        
        log.info("🚀 启动数据同步任务...");
        
        syncTask = CompletableFuture.runAsync(() -> {
            try {
                isRunning.set(true);
                
                // 创建数据流
                DataStreamSource<String> sourceStream = env
                        .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL CDC Source")
                        .setParallelism(1);
                
                // 数据处理和输出
                sourceStream
                        .map(new DataProcessor())
                        .name("Data Processor")
                        .addSink(new PrintSinkFunction<>())
                        .name("Console Sink");
                
                log.info("✅ 数据同步任务开始执行");
                env.execute("Flink CDC Sync Job");
                
            } catch (Exception e) {
                log.error("❌ 数据同步任务执行失败: {}", e.getMessage(), e);
                throw new RuntimeException(e);
            } finally {
                isRunning.set(false);
                log.info("🔚 数据同步任务结束");
            }
        });
        
        log.info("✅ 数据同步任务启动成功");
    }

    /**
     * 停止数据同步
     */
    public synchronized void stopSync() {
        if (!isRunning.get()) {
            log.warn("⚠️ 同步任务未在运行");
            return;
        }
        
        log.info("🛑 停止数据同步任务...");
        
        if (syncTask != null) {
            syncTask.cancel(true);
        }
        
        isRunning.set(false);
        log.info("✅ 数据同步任务已停止");
    }

    /**
     * 获取同步状态
     */
    public boolean isRunning() {
        return isRunning.get();
    }

    /**
     * 🔄 数据处理器
     * 实现数据一致性和迁移逻辑
     * 
     * 【数据处理段落】- CDC 数据解析和处理核心
     * 此段落实现对 MySQL CDC 数据的完整处理流程：
     * 
     * 1. JSON 数据解析：
     *    - 使用 Jackson ObjectMapper 解析 Debezium JSON 格式
     *    - 提取数据库名、表名、操作类型、时间戳等关键信息
     * 
     * 2. 操作类型映射：
     *    - c (create) -> 新增操作
     *    - u (update) -> 更新操作  
     *    - d (delete) -> 删除操作
     *    - r (read) -> 读取操作（全量同步时）
     * 
     * 3. 数据一致性检查：
     *    - 验证 CDC 数据的完整性和有效性
     *    - 检查必要字段（source、op、ts_ms）是否存在
     *    - 记录一致性检查结果用于监控
     * 
     * 4. 结果格式化：
     *    - 构建统一的输出格式便于监控和调试
     *    - 包含操作类型、表信息、时间戳等关键信息
     * 
     * 5. 错误处理：
     *    - 捕获和处理数据解析异常
     *    - 返回错误信息而不是中断整个流程
     *    - 记录详细的错误日志用于问题排查
     */
    public static class DataProcessor implements MapFunction<String, String> {
        private final ObjectMapper objectMapper = new ObjectMapper();
        
        @Override
        public String map(String jsonData) throws Exception {
            try {
                // 解析 CDC 数据
                JsonNode jsonNode = objectMapper.readTree(jsonData);
                
                // 提取关键信息
                String database = jsonNode.path("source").path("db").asText();
                String table = jsonNode.path("source").path("table").asText();
                String operation = jsonNode.path("op").asText();
                long timestamp = jsonNode.path("source").path("ts_ms").asLong();
                
                // 构建处理结果
                String result = String.format(
                    "[%s] %s.%s - %s | 时间: %d",
                    mapOperation(operation), database, table, operation, timestamp
                );
                
                // 数据一致性检查
                if (isDataConsistent(jsonNode)) {
                    log.debug("✅ 数据一致性检查通过: {}", result);
                } else {
                    log.warn("⚠️ 数据一致性检查失败: {}", result);
                }
                
                return result;
                
            } catch (Exception e) {
                log.error("❌ 数据处理失败: {}", jsonData, e);
                return "ERROR: " + e.getMessage();
            }
        }
        
        private String mapOperation(String op) {
            switch (op) {
                case "c": return "新增";
                case "u": return "更新";
                case "d": return "删除";
                case "r": return "读取";
                default: return "未知";
            }
        }
        
        private boolean isDataConsistent(com.fasterxml.jackson.databind.JsonNode jsonNode) {
            // 简单的数据一致性检查逻辑
            return jsonNode.has("source") && 
                   jsonNode.has("op") && 
                   jsonNode.path("source").has("ts_ms");
        }
    }
}